package com.zxinlee.m3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;

/**
 * RocketMQ 提供了可靠性消息，也叫事务消息
 * <p>
 * 1. 发送半消息（半消息不会发送给消费者）
 * 2. 执行本地事务
 * 3. 提交消息
 * <p>
 * 两种错误情况：
 * <p>
 * 1. 事务执行失败时回滚消息
 * 2. 服务器无法得知消息状态时，需要主动回查消息状态
 */
public class Producer {
    public static void main(String[] args) throws MQClientException {
        // 创建事务消息生产者
        TransactionMQProducer producer03 = new TransactionMQProducer("producer03");

        // 设置name server
        producer03.setNamesrvAddr("192.168.64.141:9876");

        // 设置事务消息监听器,执行本地事务,处理服务器回查
        producer03.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {

                // 执行本地事务的方法中,一般不会返回UNKNOW 状态,
                // 这里只是为了模拟网络中断,无法向服务器发送状态的情况
                // if (Math.random()<1) return LocalTransactionState.UNKNOW;

                System.out.println("执行本地事务,参数: " + arg);
                if (Math.random() < 0.5) {
                    System.out.println("本地事务执行成功");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                System.out.println("本地事务执行失败");
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("处理服务器的事务状态回查");
                System.out.println("模拟网络中断");
                return LocalTransactionState.UNKNOW;
            }
        });

        // 启动
        producer03.start();

        // 发送事务消息,会触发监听器来执行本地事务
        while (true) {
            System.out.println("输入消息:");
            String msg = new Scanner(System.in).nextLine();
            Message message = new Message("Topic03", msg.getBytes());
            producer03.sendMessageInTransaction(message, "执行本地事务的业务参数数据");
        }

    }
}
